package com.pw.gmall.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.annotation.NoSink;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.TableProcess;
import com.pw.gmall.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author: linux_future
 * @since: 2022/3/18
 **/
public class FlinkSinkUtils {


    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }

    public static <T> SinkFunction<T> getClickhouseSink(String db, String tableName, Class<T> cls) {
        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL + db;
        //insert into tableName(id..) values(..)
        StringBuffer sql = new StringBuffer();
        String keys = AtguiguUtils.fieldsTOString(cls.getDeclaredFields());
        sql.append("insert into " + tableName + "(")
                .append(keys + ") values(")
                .append(keys.replaceAll("[^,]+", "?"))
                .append(")");

        System.out.println("sql: " + sql);
        return getJdbcSink(driver, url, null, null, sql.toString());
    }

    private static <T> SinkFunction<T> getJdbcSink(String driver, String url, String username, String password, String sql) {
        return JdbcSink.sink(sql, new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T t) throws SQLException {
                        System.out.println("开始写入数据accept======= ");
                        //获取类字段
                        Class<?> tClass = t.getClass();
                        Field[] fields = tClass.getDeclaredFields();
                        try {
                            for (int i = 0, p = 1; i < fields.length; i++) {
                                Field field = fields[i];
                                NoSink noSink = field.getAnnotation(NoSink.class);
                                if (noSink == null) {
                                    field.setAccessible(true);
                                    Object vs = field.get(t);
                                    ps.setObject(p++, vs);
                                }
                            }
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withMaxRetries(3).withBatchSize(1024).withBatchIntervalMs(3000).build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(url).withDriverName(driver).withUsername(username)
                        .withPassword(password)
                        .build()
        );
    }
}
